package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

/**
 * @author Amos
 * @date 2022/4/28
 * 事务消息
 */

public class Producer {
    public static void main(String[] args) throws MQClientException {
        // 新建事务消息的生产者
        TransactionMQProducer p = new TransactionMQProducer("p3");
        // 设置name server
        p.setNamesrvAddr("192.168.126.133:9876");
        // 设置监听器，监听事务消息的发送，信息一发送，就执行本地事务
        p.setTransactionListener(new TransactionListener() {
            // 执行本地事务、执行业务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 模拟网络故障，无法通知rocketmq事务状态
                if (Math.random() < 1) { // 100%执行
                    System.out.println("模拟网络中断，无法通知rocketmq事务状态");
                    // 当前方法中，一般不会用到unknow状态
                    return LocalTransactionState.UNKNOW;
                }

                // msg: 消息对象， arg：业务数据参数
                System.out.println("在这里执行业务，业务参数: " + arg);

                if (Math.random() < 0.5) {
                    System.out.println("本地事务执行成功");
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else {
                    System.out.println("本地事务执行失败");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            // 处理 rocketmq 反向状态回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("处理rocketmq服务器的事务状态回查");
                return LocalTransactionState.UNKNOW;
            }
        });
        // 启动
        p.start();
        // 发送事务消息，会触发一个监听器来执行本地事务
        while (true) {
            System.out.print("输入消息：");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            p.sendMessageInTransaction(msg, "业务数据参数");
        }
    }
}
